package com.tang.process;

import com.tang.bean.WaterSensor;
import com.tang.functions.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 实现TopN思路1, ProcessAllWindow实现TopN
 *
 * @author tang
 * @since 2023/07/07 17:22
 */
public class ProcessWindowAllTopNDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        SingleOutputStreamOperator<WaterSensor> sensorDataStream = env.socketTextStream("192.168.70.141", 7777)
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(((element, recordTimestamp) -> element.getTs() * 1000L))
                );

        // 窗口是10秒，滑动步长是5秒，就可以实现需求
        sensorDataStream.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .process(new MyTopNProcess())
                .print();


        env.execute();
    }

    private static class MyTopNProcess extends ProcessAllWindowFunction<WaterSensor, String, TimeWindow> {

        @Override
        public void process(ProcessAllWindowFunction<WaterSensor, String, TimeWindow>.Context context,
                            Iterable<WaterSensor> elements, Collector<String> out) {
            // 定义一个hashMap用来存，key=vc, value=count值
            Map<Integer, Integer> vcCountMap = new ConcurrentHashMap<>();
            // 1.遍历数据
            for (WaterSensor element : elements) {
                if (vcCountMap.containsKey(element.getVc())) {
                    // 1.1 如果key存在，不是这个key的第一条数据，那么直接累加
                    vcCountMap.put(element.getVc(), vcCountMap.get(element.getVc()) + 1);
                } else {
                    // 1.2 key不存在，初始化
                    vcCountMap.put(element.getVc(), 1);
                }
            }

            // 2.对count值进行排序：利用List来进行排序
            List<Tuple2<Integer, Integer>> list = new ArrayList<>();
            for (Map.Entry<Integer, Integer> entry : vcCountMap.entrySet()) {
                list.add(new Tuple2<>(entry.getKey(), entry.getValue()));
            }
            // 对list进行排序，根据count值降序
            list.sort((o1, o2) -> {
                // 降序，后 减 前
                return o2.f1 - o1.f1;
            });

            // 3.取出count最大的两个vc
            StringBuilder outBuilder = new StringBuilder();
            outBuilder.append("=========================\n");
            // 3.1 遍历排序后的list，取出前两个，考虑可能list只有一个的情况，这里还是取个list size和2的最小。
            for (int i = 0, len = Math.min(2, list.size()); i < len; i++) {
                Tuple2<Integer, Integer> vcTuple = list.get(i);
                outBuilder.append("Top").append(i + 1).append("\n")
                        .append("vc = ").append(vcTuple.f0).append("\n")
                        .append("count = ").append(vcTuple.f1).append("\n")
                        .append("窗口结束时间 = ")
                        .append(DateFormatUtils.format(context.window().getEnd(),
                                "yyyy-MM-dd HH:mm:ss.SSS"))
                        .append("\n")
                        .append("=========================\n");
            }
            out.collect(outBuilder.toString());

        }
    }

}
